package com.boyi.shared.kafka;

import com.boyi.shared.util.SerializeUtil;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

public class KafkaObjectDeseralizer implements Deserializer<Object> {

	@Override
	public void configure(Map<String, ?> configs, boolean isKey) {
	}

	@Override
	public Object deserialize(String topic, byte[] bytes) {
		if (bytes == null) {
			return null;
		}
		Object obj = null;
		try {
			if (bytes != null) {
				obj = SerializeUtil.deserialize(bytes);
			}
			return obj;
		} catch (IOException e) {
			throw new SerializationException("Can't deserialize data [" + Arrays.toString(bytes) +
					"] from topic [" + topic + "]", e);
		} catch (ClassNotFoundException e) {
			throw new SerializationException("Can't deserialize data [" + Arrays.toString(bytes) +
					"] from topic [" + topic + "]" + " due to " + obj.getClass() + " not found", e);
		}
	}

	@Override
	public void close() {

	}
}
